//
// Created by wuzhe on 2022/4/22.
//

#include <string>
#include <memory>
#include "RedisAsyncUtils.h"
#include "Reactor.hh"
#ifdef __linux__
#include <sys/epoll.h>
#else
#include <sys/event.h>
#endif
extern struct JJ::myevent_s g_events[MAX_EVENTS+ 1 + 1];   /* +1 daoshu dier ge 用于 listen fd */ /* zuihouyige redis */

int RedisAsync::send(const uint8_t *sendBuf, int sendBufLen) {
    std::string con((char*)sendBuf, sendBufLen);
    int ret = ::send(fd, sendBuf, sendBufLen, 0);
    if (ret < 0) {
        printf("send error : %d\n", errno);
        return -3;
    }
    return 0;
}

int RedisAsync::set(std::string key, std::string value, std::function<void(std::string&)> c) {
    char* cmdBuf = NULL;
    int len = 0;

    len = redisFormatCommand(&cmdBuf, "SET %s %s", key.data(), value.data());
    if (len < 0) {
        printf("redisFormatCommand error! len = %d\n", len);
        return -1;
    }
//    printf("redisFormatCommand result: len = %d, cmdBuf = %s\n", len, cmdBuf);
    int ret = 0;
    ret = send((const uint8_t*)cmdBuf, len);
    if (ret) {
        printf("send error:%d!\n", ret);
        free(cmdBuf);
        cmdBuf = NULL;
        return -2;
    }
    closureQueue.push_front(c);
    return 0;
}

int RedisAsync::get(std::string key, std::function<void(std::string &)> c) {
    char* cmdBuf = NULL;
    int len = 0;

    len = redisFormatCommand(&cmdBuf, "GET %s", key.data());
    if (len < 0) {
        printf("redisFormatCommand error! len = %d\n", len);
        return -1;
    }
//    printf("redisFormatCommand result: len = %d, cmdBuf = %s\n", len, cmdBuf);
    int ret = 0;
    char recvBuf[1024] = {0};
    int recvBufLen = sizeof(recvBuf);
    ret = send((const uint8_t*)cmdBuf, len);
    if (ret) {
        printf("send error:%d!\n", ret);
        free(cmdBuf);
        cmdBuf = NULL;
        return -2;
    }
    closureQueue.push_front(c);
    return 0;
}

int RedisAsync::del(std::string key, std::function<void(std::string &)> c) {
    char* cmdBuf = NULL;
    int len = 0;

    len = redisFormatCommand(&cmdBuf, "DEL %s", key.data());
    if (len < 0) {
        printf("redisFormatCommand error! len = %d\n", len);
        return -1;
    }
//    printf("redisFormatCommand result: len = %d, cmdBuf = %s\n", len, cmdBuf);
    int ret = 0;
    char recvBuf[1024] = {0};
    int recvBufLen = sizeof(recvBuf);
    ret = send((const uint8_t*)cmdBuf, len);
    if (ret) {
        printf("send error:%d!\n", ret);
        free(cmdBuf);
        cmdBuf = NULL;
        return -2;
    }
    closureQueue.push_front(c);
    return 0;
}


int RedisAsync::find(std::string key, std::function<void(std::list<std::string>&)> f) {
    auto aptr = std::make_shared<std::function<void(std::list<std::string>&)>>(f);

    auto c = [&, aptr](std::string &str) {
        char* a = strtok(str.data(), "\n");
        std::list<std::string> res;
        while(a != nullptr) {
            res.emplace_back(a);
            a = strtok(NULL, "\n");
        }
        (*aptr)(res);
    };
    char* cmdBuf = NULL;
    int len = 0;

    len = redisFormatCommand(&cmdBuf, "KEYS %s", key.data());
    if (len < 0) {
        printf("redisFormatCommand error! len = %d\n", len);
        return -1;
    }
//    printf("redisFormatCommand result: len = %d, cmdBuf = %s\n", len, cmdBuf);
    int ret = 0;
    char recvBuf[1024] = {0};
    int recvBufLen = sizeof(recvBuf);
    ret = send((const uint8_t*)cmdBuf, len);
    if (ret) {
        printf("send error:%d!\n", ret);
        free(cmdBuf);
        cmdBuf = NULL;
        return -2;
    }
    closureQueue.push_front(c);
    return 0;
}

bool RedisAsync::connect(std::string host, int port) {

    fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) {
        printf("socket error : %d\n", errno);
        return -1;
    }

    struct sockaddr_in addr;
    memset(&addr, 0x00, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(6379);
    addr.sin_addr.s_addr = inet_addr("127.0.0.1");

    int ret = ::connect(fd, (struct sockaddr*)&addr, sizeof(addr));
    if (ret) {
        printf("connect error : %d\n", errno);
        close(fd);
        return false;
    }

    return true;
}

void redisRecv(int lfd, int, void *) {
    char buf[BUFSIZ];
    int ret = recv(lfd, buf, BUFSIZ, 0);
    if (ret <= 0) {
        printf("recv error : %s\n", strerror(errno));
        return;
    }
    auto redis = RedisAsync::getSingleton();
    if (BUFSIZ - redis->readRightIndex >= ret) {
        memcpy(redis->recvBuf + redis->readRightIndex, buf, ret);
    } else if (ret - (BUFSIZ - redis->readRightIndex) < redis->readLeftIndex) {
        int len1 = BUFSIZ - redis->readRightIndex;
        int len2 = ret - BUFSIZ - redis->readRightIndex;
        memcpy(redis->recvBuf + redis->readRightIndex, buf, len1);
        memcpy(redis->recvBuf, buf + len1, len2);
    }
    redis->readRightIndex = (redis->readRightIndex + ret) % BUFSIZ;
}

void RedisAsync::processRecv() {
    // 读缓冲空
    if (this->readLeftIndex == this->readRightIndex)
        return;
    char *content = nullptr;
    int len = 0;
    if (this->readLeftIndex < this->readRightIndex) {
        len = this->readRightIndex - this->readLeftIndex;
        content = (char *) malloc(len * sizeof(char));
        memcpy(content, this->recvBuf + this->readLeftIndex, len);
    } else {
        int len1 = BUFSIZ - this->readLeftIndex;
        int len2 = this->readRightIndex;
        len = len1 + len2;
        content = (char *) malloc(len * sizeof(char));
        memcpy(content, this->recvBuf + this->readLeftIndex, len1);
        memcpy(content + len1, this->recvBuf, len2);
    }
    int index = 0;
    processResp(content, len, index);
    // empty buf
    free(content);
    this->readLeftIndex = (this->readLeftIndex + index) % BUFSIZ;

}

void RedisAsync::processResp(char * content, int len, int& index) {
    index = 0;
    redisReader *reader = NULL;
    void *reply = NULL;

    reader = redisReaderCreate();
    if (reader == NULL) {
        printf("redisReaderCreate error!\n");
        return;
    }
    redisReaderFeed(reader, content, len);

    while (true) {
        int ret = redisReaderGetReply(reader, &reply);
        if (ret != REDIS_OK) {
            printf("redisReaderGetReply error:%d!\n", ret);
            redisReaderFree(reader);
            reader = NULL;
            break;
        }
        if (reply == NULL) {
//            printf("redisReply is NULL!\n");
            redisReaderFree(reader);
            reader = NULL;
            break;
        }
        index = len - reader->len + reader->pos;

        struct redisReply *r = (struct redisReply*)reply;
//        printf("r->type = %d\n", r->type);
        if (REDIS_REPLY_ERROR == r->type) {
            printf("redisReply return REDIS_REPLY_ERROR!\n");
            freeReplyObject(reply);
            closureQueue.pop_back();
            break;
        }
        std::string str{};

        if (REDIS_REPLY_STRING != r->type) {

            if (REDIS_REPLY_NIL == r->type) {
                str += "nil";
            }
            else if (REDIS_REPLY_ARRAY == r->type) {
                for (int k = 0; k < r->elements; ++k) {
                    redisReply* reply = r->element[k];
                    str += reply->str;
                    str += "\n";
                }
            }
        } else {
            str += r->str;
        }
        auto f = closureQueue.back();
        closureQueue.pop_back();
        f(str);
        //

    }
    freeReplyObject(reply);
}

void RedisAsync::initSocket(int efd) const {
    eventset(&g_events[MAX_EVENTS + 1], this->fd,true, redisRecv, &g_events[MAX_EVENTS + 1]);
#ifdef __linux__
    eventadd(efd, EPOLLIN, &g_events[MAX_EVENTS + 1]);
#else
    eventadd(efd, EVFILT_READ, &g_events[MAX_EVENTS + 1]);

#endif
}

